package sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
import org.apache.spark.sql.types.{DataType, LongType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object SparkSql2 {
  def main(args:Array[String]):Unit={
    val sparkconf=new SparkConf().setMaster("local[*]").setAppName("sparkSql")
    val spark=SparkSession.builder().config(sparkconf).getOrCreate()
    val df: DataFrame = spark.read.json("data/user.json")
    df.createOrReplaceTempView("user")
    //自定义一个函数加入到spark中
    spark.udf.register("ageAVG",new MyAvgUDF())
    spark.sql("select ageAVG(age) from user").show()

    spark.close()
  }
  //自定义聚合函数：计算年龄的平均值
  class MyAvgUDF extends UserDefinedAggregateFunction{
    //输入的数据结构
    override def inputSchema: StructType = {
        StructType(
          Array(
            StructField("age",LongType)
          )
        )
    }
    //缓冲区的数据结构
    override def bufferSchema: StructType = {
      StructType(
        Array(
          StructField("total",LongType),
          StructField("count",LongType)
        )
      )
    }
   //函数计算结果的数据类型Out
    override def dataType: DataType = LongType
   //函数的稳定性
    override def deterministic: Boolean = true
   //缓冲区初始化
    override def initialize(buffer: MutableAggregationBuffer): Unit = {
//      buffer(0)=0L
//      buffer(1)=0L
      buffer.update(0,0L)
      buffer.update(1,0L)
    }
//根据输入的值更新缓冲区
    override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
      buffer.update(0,buffer.getLong(0)+input.getLong(0))
      buffer.update(1,buffer.getLong(1)+1)
    }
  //缓冲区数据合并
    override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
      buffer1.update(0,buffer1.getLong(0)+buffer2.getLong(0))
      buffer1.update(1,buffer1.getLong(1)+buffer2.getLong(1))
    }

    override def evaluate(buffer: Row): Any = {
      buffer.getLong(0)/buffer.getLong(1)
    }
  }

}

